/**
 * Copyright (C), 2015-2018, XXX有限公司
 * FileName: JdbcInsertOrUpdateBolt
 * Author:   An-Il
 * Date:     2018/11/15 9:48
 * Description: 整合jdbc查询或更新信息（以wordcount为例）
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
package com.blog.storm.example.integration.jdbc;

import org.apache.commons.lang.Validate;
import org.apache.storm.jdbc.bolt.AbstractJdbcBolt;
import org.apache.storm.jdbc.bolt.JdbcLookupBolt;
import org.apache.storm.jdbc.common.Column;
import org.apache.storm.jdbc.common.ConnectionProvider;
import org.apache.storm.jdbc.mapper.JdbcLookupMapper;
import org.apache.storm.jdbc.mapper.JdbcMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Types;
import java.util.ArrayList;
import java.util.List;

/**
 * 〈一句话功能简述〉<br>
 * 〈整合jdbc查询或更新信息（以wordcount为例）〉
 *
 * @author An-Il
 * @create 2018/11/15
 * @since 1.0.0
 */
public class JdbcInsertOrUpdateBolt extends AbstractJdbcBolt {
    private static final Logger LOG = LoggerFactory.getLogger(JdbcLookupBolt.class);

    private String insertQuery;

    private String updateQuery;

    private String selectQuery;

    private JdbcLookupMapper jdbcLookupMapper;

    private JdbcMapper jdbcInsertMapper;

    public JdbcInsertOrUpdateBolt(ConnectionProvider connectionProvider, String selectQuery, JdbcLookupMapper jdbcLookupMapper) {
        super(connectionProvider);

        Validate.notNull(selectQuery);
        Validate.notNull(jdbcLookupMapper);

        this.selectQuery = selectQuery;
        this.jdbcLookupMapper = jdbcLookupMapper;
    }

    public JdbcInsertOrUpdateBolt withInsertQueryAndInsertMapper(String insertQuery, JdbcMapper jdbcInsertMapper) {
        Validate.notNull(jdbcInsertMapper);
        Validate.notNull(insertQuery);
        this.insertQuery = insertQuery;
        this.jdbcInsertMapper = jdbcInsertMapper;
        return this;
    }

    public JdbcInsertOrUpdateBolt withUpdateQuery(String updateQuery) {
        Validate.notNull(updateQuery);
        this.updateQuery = updateQuery;
        return this;
    }

    public JdbcInsertOrUpdateBolt withQueryTimeoutSecs(int queryTimeoutSecs) {
        this.queryTimeoutSecs = queryTimeoutSecs;
        return this;
    }

    @Override
    protected void process(Tuple tuple) {
        try {
            List<Column> queryColumns = jdbcLookupMapper.getColumns(tuple);
            List<List<Column>> result = jdbcClient.select(this.selectQuery, queryColumns);
            //如果数据库存在数据，则更新数据
            if (result != null && result.size() != 0) {
                for (List<Column> row : result) {
                    List<Values> values = jdbcLookupMapper.toTuple(tuple, row);
                    //更新数据
                    for (Values value : values) {
                        value.set(1, (Integer) value.get(1) + 1);
                        this.update(value);
                        LOG.debug("update data" + tuple.toString());
                    }
                }
            } else {
                //插入数据
                List<List<Column>> columnLists = new ArrayList<List<Column>>();
                List<Column> insertColumns = jdbcInsertMapper.getColumns(tuple);
                columnLists.add(insertColumns);
                this.jdbcClient.executeInsertQuery(this.insertQuery, columnLists);
                LOG.debug("insert data:" + tuple.toString());
            }
            this.collector.ack(tuple);
        } catch (Exception e) {
            this.collector.reportError(e);
            this.collector.fail(tuple);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        jdbcLookupMapper.declareOutputFields(outputFieldsDeclarer);
    }

    private void update(Values value) throws Exception {
        List<List<Column>> columnLists = new ArrayList<List<Column>>();
        List<Column> updateColumns = new ArrayList<Column>();
        updateColumns.add(new Column("count",value.get(1), Types.INTEGER));
        updateColumns.add(new Column("word",value.get(0), Types.VARCHAR));
        columnLists.add(updateColumns);
        //executeInsertQuery改造下,注意順序是count，word
        this.jdbcClient.executeInsertQuery(this.updateQuery, columnLists);
    }
}